/*
 * (c) Copyright 2008, 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 * [See end of file]
 */

package com.hp.hpl.jena.clusteredtdb.comms;

import static java.lang.String.format;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/** Unit of network transfer.
 * 
 * @author Andy Seaborne
 * @version $Id$
 */
public class NetBlock
{
    private static Logger log = LoggerFactory.getLogger(NetBlock.class) ;
    static int HeaderSize = 4 ;
    ByteBuffer block ;
    ByteBuffer data ;
    boolean inUse = true ;
    
    public static NetBlock alloc(int size)
    {
        // Later - recycle blocks?
        return new NetBlock(size) ;
    }
    
    public static void write(WritableByteChannel channel, NetBlock netBlock)
    {
        if ( ! netBlock.inUse )
            throw new CommsException("write: Block has been released") ;
        
        try
        {
            ByteBuffer net = netBlock.getNetBuffer() ;
            log.debug(format("Write: %d", netBlock.dataSize())) ;
            channel.write(net) ;
        } catch (IOException ex)
        {
            throw new CommsException(ex) ;
        }
    }
    
    public static NetBlock read(ReadableByteChannel channel)
    {
//        if ( ! netBlock.inUse )
//            throw new CommsException("read: Block has been released") ;

        try
        {
            // Silly, simple version. Fixed size (fixed max size) expected blocks is better. 
            ByteBuffer bb = ByteBuffer.allocate(HeaderSize) ;

            readExactly(channel, bb, 4) ;       // Get EOF here

            int bytes = bb.getInt(0) ;          // Size without header
            bb = ByteBuffer.allocate(bytes+HeaderSize) ;
            // Fake a block
            bb.putInt(bytes) ;
            readExactly(channel, bb, bytes) ;
//            bb.flip();
//            bb.limit(bytes) ;
            
            NetBlock netBlock = new NetBlock() ;
            netBlock.block = bb ;
            bb.position(4) ;
            netBlock.data = bb.slice();
            return netBlock ;
        } catch (IOException ex)
        {
            return null ;
        }
        
        
    }
    
    private static void readExactly(ReadableByteChannel channel, ByteBuffer bb, int len) throws IOException
    {
        if ( len == 0 )
        {
            log.debug("Zero read request") ;
            return ;
        }
        
        int soFar = 0 ;
        while(true)
        {
            int x = channel.read(bb) ;
            log.debug(format("Read: %d", x)) ;
            if ( x < 0 )
                throw new CommsException("Premature end of channel") ;
            soFar += x ;
            if ( soFar == len )
                break ;
        }
        bb.position(0) ;
        
    }
    
    private NetBlock()
    {
        inUse = true ;
    }
    
    /** Maximum size of the data */
    private NetBlock(int size)
    {
        this() ;
        int len = size+HeaderSize ;
        block = allocByteBuffer(len) ;
        //block.order(Const.NetworkOrder) ;
        block.order(ByteOrder.BIG_ENDIAN) ;
        block.position(HeaderSize) ;
        data = block.slice() ;
    }
    
    public ByteBuffer getBuffer()
    {
        return data ;
    }
    
    public int dataSize()
    {
        return data.capacity() ;
    }
    
    public void release()
    {
        inUse = false ;
    }
    
    private ByteBuffer getNetBuffer()
    {
        int x = data.limit() ;
        int size = data.limit() ;
        block.putInt(0, size) ;     // Message bytes - excludes header.
        block.limit(size+HeaderSize) ;
        block.position(0) ;
        // c.f. flip.
        return block ;
    }
    
    private ByteBuffer allocByteBuffer(int len)
    {
        return ByteBuffer.allocate(len) ;
        // Direct is a bit slower (20%? windows) - but could be pooled and managed. 
        //return ByteBuffer.allocateDirect(len) ;
    }
}

/*
 * (c) Copyright 2008, 2009 Hewlett-Packard Development Company, LP
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. The name of the author may not be used to endorse or promote products
 *    derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
 * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
